#include <stdafx.h>

#ifndef WINDOWS
#include <errno.h>
#endif

#include "Socket.h"
#include "PC2DataStore.h"

#include "PC2Node.h"

namespace PC2
{

template<int Size>
struct FDSet
{
	u_int fd_count;
	SOCKETTYPE  fd_array[Size];
};

PC2Node::PC2Node(NodeInfo const& aInfo, std::vector<NodeInfo> const& aParticipantIDSet, char const* apDBPath)
	: m_Info(aInfo)
	, m_pSocket(new Socket())
	, m_pStore(NULL)
	, m_Parts()
{
	for (auto it = aParticipantIDSet.begin(); it != aParticipantIDSet.end();it++)
	{
		m_Parts[it->m_Id] = *it;
	}

	if(NULL != apDBPath)
		m_pStore = new PC2DataStore(apDBPath);

	m_pSocket->SetNonBlockMode();
	m_pSocket->Bind(aInfo.m_IP.c_str(), aInfo.m_Port, false);
//	m_pSocket->SetReceiveBufferSize(32768);
}

PC2Node::PC2Node( NodeInfo const& aInfo )
	: m_Info(aInfo)
	, m_pSocket(new Socket())
	, m_pStore(NULL)
	, m_Parts()
{
	m_pSocket->SetNonBlockMode();
	m_pSocket->Bind(aInfo.m_IP.c_str(), aInfo.m_Port, false);
//	m_pSocket->SetReceiveBufferSize(32768);
}


PC2Node::~PC2Node()
{
	delete m_pStore;
	delete m_pSocket;
}

bool PC2Node::WaitForData( Trans* apTrans, std::string* apData, int32_t aTimeout, char* apSrcAddr, unsigned short* apSrcPort )
{
#ifdef WINDOWS
	FDSet<2> lSet;
	lSet.fd_count = 1;
	lSet.fd_array[0] = m_pSocket->GetNativeHandle();
#else
	fd_set lSet;
	FD_SET(m_pSocket->GetNativeHandle(), &lSet);
#endif


	char lBuffer[1024];

	timeval lTime;
	lTime.tv_sec = 0;
	lTime.tv_usec = aTimeout*1000;

	while (true)
	{
		int lCount = select(m_pSocket->GetNativeHandle() + 1, (fd_set*)&lSet, NULL, NULL, &lTime);
		if(lCount > 0)
		{
			int lLen;
			if(NULL == apSrcAddr)
				lLen = m_pSocket->Receive(lBuffer, 1024, 0);
			else
				lLen = m_pSocket->ReceiveFrom(lBuffer, 1024, 0, apSrcAddr, apSrcPort);
			if(lLen == SOCKET_ERROR)
			{
#ifdef WINDOWS
				int lError = WSAGetLastError();
#else
				int lError = errno;
#endif
				if(0 == lError)
				{
					continue;
				}
#ifdef WINDOWS
				else if(WSAECONNRESET == lError)
				{
					//windows bug http://support.microsoft.com/kb/263823
					continue;
				}
#endif
				else
				{
					LOG("socket receive get error %d\n", lError);
					return false;
				}
			}
			*apTrans = *(Trans*)lBuffer;
			if(lLen > sizeof(Trans))
			{
				apData->assign(lBuffer + sizeof(Trans), lLen - sizeof(Trans));
			}
			else
			{
				apData->clear();
			}

//			LOG("received trans node_id %u trans_id %u type %u size %u\n", apTrans->m_NodeId, apTrans->m_TransId, apTrans->m_Type, apTrans->m_Size);

			return true;
		}
		else
		{
			return false;
		}
	}

}

void PC2Node::SendTrans( NodeInfo* apNode, Trans* apTrans )
{
	int lRes = m_pSocket->SendTo(apNode->m_IP.c_str(), apNode->m_Port, (char const*)apTrans, apTrans->m_Size, 0);

//	LOG("sent trans node_id %u trans_id %u type %u size %u\n", apNode->m_Id, apTrans->m_TransId, apTrans->m_Type, apTrans->m_Size);

	if((uint32_t)lRes != apTrans->m_Size)
	{
		LOG("error Send_trans socket return %d\n", lRes);
	}
}

PC2DataStore* PC2Node::GetDataStore()
{
	return m_pStore;
}

void PC2Node::HandleDecisionRequest( TransactionInfoSet* apInfoSet, Trans* apReceivedTrans, char const* apRemoteIP, unsigned short aPort )
{
	TransactionInfoSet lRelatedSet;
	if(NULL == apInfoSet || apInfoSet->m_LastTrans.m_TransId != apReceivedTrans->m_TransId)
	{
		if(m_pStore->LoadTranInfoSet(apReceivedTrans->m_TransId, &lRelatedSet, false))
		{
			apInfoSet = &lRelatedSet;
		}
		else
		{
			LOG("received decision request while result not in history %u\n", apReceivedTrans->m_TransId);
			return;
		}
	}

	NodeInfo lNode;
	lNode.m_IP = apRemoteIP;
	lNode.m_Port = aPort;

	Trans lResponse;
	lResponse.m_NodeId = m_Info.m_Id;
	lResponse.m_Size = sizeof(Trans);
	lResponse.m_TransId = apReceivedTrans->m_TransId;
		

	if(EnumCommitRequest == apInfoSet->m_LastTrans.m_Type || EnumCommitAck == apInfoSet->m_LastTrans.m_Type)
	{
		lResponse.m_Type = EnumCommitRequest;
		SendTrans(&lNode, &lResponse);
	}
	else if(EnumAbortRequest == apInfoSet->m_LastTrans.m_Type || EnumAbortAck == apInfoSet->m_LastTrans.m_Type)
	{
		lResponse.m_Type = EnumAbortRequest;
		SendTrans(&lNode, &lResponse);
	}
}



}
